package com.shujia.flink.core;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Demo1WordCount {

    public static void main(String[] args) throws Exception {
        //1、创建flink的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度,一个并行度对应一个task
        env.setParallelism(2);

        //修改数据从上游发送到下游的缓存时间
        env.setBufferTimeout(2000);


        /*
         * 无界流
         */
        //2、读取数据
        //nc -lk 8888
        DataStream<String> linesDS = env.socketTextStream("master", 8888);


        //一行转换成多行
        DataStream<String> wordsDS = linesDS
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String line, Collector<String> out) throws Exception {
                        for (String word : line.split(",")) {
                            //将数据发送到下游
                            out.collect(word);
                        }
                    }
                });

        //转换成kv格式
        DataStream<Tuple2<String, Integer>> kvDS = wordsDS
                .map(new MapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String word) throws Exception {
                        //返回一个二元组
                        return Tuple2.of(word, 1);
                    }
                });

        //按照单词进行分组
        //底层是hash分区
        KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS
                .keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> kv) throws Exception {
                        return kv.f0;
                    }
                });

        //统计数量
        DataStream<Tuple2<String, Integer>> countDS = keyByDS
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> kv1,
                                                          Tuple2<String, Integer> kv2) throws Exception {
                        int count = kv1.f1 + kv2.f1;
                        return Tuple2.of(kv1.f0, count);
                    }
                });

        //打印结果
        countDS.print();

        //3、启动flink
        env.execute("wc");
    }
}

